package cn.itcast.tags.ml.regression

import org.apache.spark.ml.linalg
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.regression.{LinearRegression, LinearRegressionModel, LinearRegressionTrainingSummary}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.storage.StorageLevel

/**
 * @author: xu
 * @desc: 使用线性回归算法，对波士顿房价数据集构建回归模型，预测和评估模型
 *        - 波士顿房价数据集共506条，13个属性（特征值features），1个房价（预测值target）
 */
object BostonLinearRegression {
  def main(args: Array[String]): Unit = {
    // 构建SparkSession实例对象，通过建造者模式创建
    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .master("local[3]")
      .config("spark.sql.shuffle.partitions", "3")
      .getOrCreate()
    import spark.implicits._

    // TODO: 1. 读取原始数据集：波士顿房价数据集
    val bostonHousePriceDS: Dataset[String] = spark.read
      .textFile("datas/housing/housing.data")
      .filter(line => line != null && line.trim.split("\\s+").length == 14)

    // TODO: 2. 获取特征features和标签label
    val bostonDF: DataFrame = bostonHousePriceDS
      .mapPartitions { iter =>
        iter.map { line =>
          val parts: Array[String] = line.trim.split("\\s+")
          // 获取最后一列房价标签
          val label: Double = parts(parts.length - 1).toDouble
          // 获取前面13个特征features
          val doubles: Array[Double] = parts.dropRight(1).map(_.toDouble)
          val features: linalg.Vector = Vectors.dense(doubles)
          (features, label)
        }
      }
      .toDF("features", "label")
    // bostonDF.printSchema()
    // bostonDF.show(10, truncate = false)

    // TODO: 将特征数据集划分为训练集和测试集，并将训练缓存
    val Array(trainingDF, testingDF) = bostonDF.randomSplit(Array(0.8, 0.2))
    trainingDF.persist(StorageLevel.MEMORY_AND_DISK).count()

    // TODO: 3. 将特征进行各种转换，比如正则化、归一化、标准化等
    // 由于线性回归算法，默认情况下，对特征features数据进行标准化处理转换，所以此处不再进行处理

    // TODO: 4. 创建线性回归算法对象，设置参数，并且使用数据训练模型
    val lr: LinearRegression = new LinearRegression()
      // 设置特征和标签列名称
      .setFeaturesCol("features")
      .setLabelCol("label")
      // 是否对特征数据，进行标准化转换处理
      .setStandardization(true)
      // 设置算法底层的求解方式，要么最小二乘法（normal），或拟牛顿法（l-bfgs）
      .setSolver("auto")
      // 设置算法相关的超参数值
      .setMaxIter(20)
      .setRegParam(0.3) // 正则化参数
      .setElasticNetParam(0.8) // 弹性网络参数：L1正则和L2正则联合使用
    val lrModel: LinearRegressionModel = lr.fit(trainingDF)

    // TODO: 5. 模型评估
    // 打印逻辑回归的系数和截距，就是k和b
    println(s"Coefficients: ${lrModel.coefficients} Intercept: ${lrModel.intercept}")
    // 在训练集上总结模型并打印出一些指标
    val trainingSummary: LinearRegressionTrainingSummary = lrModel.summary
    println(s"numIterations: ${trainingSummary.totalIterations}")
    println(s"objectiveHistory: [${trainingSummary.objectiveHistory.mkString(",")}]")
    // sum((真实价格-预测价格)^2)/count也就是均方误差
    println(s"RMSE: ${trainingSummary.rootMeanSquaredError}")
    println(s"r2: ${trainingSummary.r2}")

    // TODO: 6. 模型模型预测
    lrModel.transform(testingDF).show(20, truncate = false)

    spark.stop()
  }
}
